package spark.example

import org.apache.spark.sql.{SparkSession,DataFrame}


object globalSort {
  def main (args: Array[String]) {
    val spark = SparkSession.builder().appName("globalSort example").master("yarn").getOrCreate()
    val input1 = args(0)
    val input2 = args(1)
    val output = args(2)
    
    val contextDF1 = readContextFromFile(spark,input1)
    val contextDF2 = readContextFromFile(spark,input2)
    // 这里一定要执行这条语句，否则无法认识 sort($"key"), "$" is not StringContext member
    import spark.implicits._

    contextDF1.union(contextDF2).sort($"key")
      // 这个map的作用是将两列转换成1列，key,value 以 "\t" 分割
      .map(x => x(0).toString + "\t" + x(1))
      // coalesce(1) 将 DataFrame 的 partitions 变成 1，主要是数据量比较小，没必要生成那么多小文件，注意如果没有这条语句将生产102个文件，其中有1个为空文件，因为总共是101行数据
      .coalesce(1)
      // DataFrameWriter 这个类主要是将DataSet 保存至存储系统，可以支持丰富的格式
      .write
      .text(output)

  }
  
case class Context(key: Int,value: String)
  def readContextFromFile(spark: SparkSession,fielPath: String) : DataFrame = {
    // 这里没有这条语句无法转调用 toDF
    import spark.implicits._
    spark.sparkContext
      .textFile(fielPath)
      .map(_.split("\t"))
      // 这里利用了 spark 的反射机制，将 rdd 转换成 DataFrame，并且创建了 schema。读取字段的类型，默认是String，第一列希望是Int类型
      .map( x => Context(x(0).toInt,x(1)))
      .toDF
  }
}

/*
提交集群的方式：
#!bin/bash
hadoop fs -rm -r hdfs://master:9000/spark-sort-global
cd /data/project/spark/spark_workstation
/data/sbt/bin/sbt compile && /data/sbt/bin/sbt package &&
/data/spark/bin/spark-submit --master yarn \
    --deploy-mode cluster \
	--num-executors 2 \
	--executor-memory '1024m' \
	--executor-cores 1 \
	--class spark.example.globalSort ./target/scala-2.11/mergedata_2.11-2.2.1.jar \
    hdfs://master:9000/a.txt \
    hdfs://master:9000/b.txt \
    hdfs://master:9000/spark-sort-global
*/

/*
hdfs://master:9000/a.txt 内容:
第一列奇数，第二列相同，测试数据随便弄的，总共50条
1	hadoop
3	hadoop
5	hadoop
7	hadoop
9	hadoop
11	hadoop
13	hadoop
15	hadoop
17	hadoop
19	hadoop

hdfs://master:9000/b.txt 内容:
第一列偶数，第二列相同，测试数据随便弄的，总共50条
0	java
2	java
4	java
6	java
8	java
10	java
12	java
14	java
16	java
18	java

最后输出的效果 第1列：0-100，第2列java/hadoop 交替
0	java
1	hadoop
2	java
3	hadoop
4	java
5	hadoop
6	java
7	hadoop
8	java
9	hadoop
10	java

*/